package cn.doitedu.kafka.day01

import java.util.Properties

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer

/**
 * 使用scala编写Kafka的生产者
 */
object ProducerDemo1 {

  def main(args: Array[String]): Unit = {


    //指定生产者的参数
    val properties = new Properties()
    //broker的地址
    // 连接kafka节点
    properties.setProperty("bootstrap.servers", "node-1.51doit.cn:9092,node-2.51doit.cn:9092,node-3.51doit.cn:9092")
    //指定key序列化方式
    properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    //指定value序列化方式
    properties.setProperty("value.serializer", classOf[StringSerializer].getName) // 两种写法都行

    //topic名称
    val topic = "test"
    //创建Kafka生产者
    val producer = new KafkaProducer[String, String](properties)

    //将数据封装到ProducerRecord中
    val record1 = new ProducerRecord[String, String](topic, "hello spark 11666")
    val record2 = new ProducerRecord[String, String](topic, "hello spark 11777")

    producer.send(record1)
    //producer.flush()
    producer.send(record2)
    //producer.flush()
    println("数据已经发送！")

    //producer.close()





  }

}
